package com.hyts.distlock.facade.zookeeper;

import com.hyts.distlock.config.LockConfig;
import com.hyts.distlock.handler.ZkOriginHandler;
import com.hyts.distlock.lock.ZookeeperLock;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * project-name:coffice-toolbox
 * package-name:com.hyts.distlock.facade.zookeeper
 * author:Libo/Alex
 * create-date:2019-05-11 09:23
 * copyright:libo-hyts-github
 * email:libo2dev@aliyun.com
 * description:此类主要用于: zookeeper 客户端操作
 */
public class ZkOrginalLock extends ZookeeperLock<ZooKeeper> {

    /**
     * 日志
     */
    Logger logger = LoggerFactory.getLogger(ZkOrginalLock.class);


    String currentLock;


    String rootLockPath;

    CountDownLatch countDownLatch;

    String waitLock;

    /**
     * Lock对象构造器
     *
     * @param lockObjectClient
     */
    public ZkOrginalLock(ZooKeeper lockObjectClient) {
        super(lockObjectClient);
        try {
            lockObjectClient.getData(LockConfig.LOCK_METADATA,false, new Stat());
            if (this.countDownLatch != null) {
                this.countDownLatch.countDown();
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public boolean tryLock(String lockName, Long timeout) {
        if(lock(lockName,timeout)){
            return true;
        }
        try {
            return waitForLock(waitLock,timeout);
        } catch (KeeperException e) {
            e.printStackTrace();
            return false;
        } catch (InterruptedException e) {
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public boolean tryLock(String lockName) {
        if(lock(lockName)){
            return true;
        }
        try {
            return waitForLock(waitLock,LockConfig.ZOOKEEPER_CONNECTION_TIMEOUT);
        } catch (KeeperException e) {
            e.printStackTrace();
            return false;
        } catch (InterruptedException e) {
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public boolean unlock(String lockName) {
        try {
            getLockObjectClient().delete(this.currentLock, -1);
        } catch (InterruptedException e) {
            e.printStackTrace();
            return false;
        } catch (KeeperException e) {
            e.printStackTrace();
            return false;
        }finally {
            try {
                getLockObjectClient().close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return true;
    }

    // 等待锁
    private boolean waitForLock(String prev, long waitTime) throws KeeperException, InterruptedException {
        Stat stat = getLockObjectClient().exists(rootLockPath + "/" + prev, true);
        if (stat != null) {
            System.out.println(Thread.currentThread().getName() + "等待锁 " + rootLockPath + "/" + prev);
            this.countDownLatch = new CountDownLatch(1);
            // 计数等待，若等到前一个节点消失，则precess中进行countDown，停止等待，获取锁
            this.countDownLatch.await(waitTime, TimeUnit.SECONDS);
            this.countDownLatch = null;
            System.out.println(Thread.currentThread().getName() + " 等到了锁");
        }
        return true;
    }

    /**
     * @param lockName
     * @param timeout
     * @return
     */
    private boolean lock(final String lockName,Long timeout){
        try {
            String currentLock =
                    getLockObjectClient().create(LockConfig.LOCK_ZOOKEEPER_PREFIX+
                    "/"+lockName+LockConfig.LOCK_SPLIT_TOKEN,new byte[0],
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            List<String> lockList = getLockObjectClient().getChildren(rootLockPath,false);
            if(lockList != null && lockList.size() > 1){
                lockList = lockList.parallelStream().map(param->param.split(LockConfig.LOCK_SPLIT_TOKEN)[0]).
                        filter(param3->param3.equals(lockName)).collect(Collectors.toList());
                Collections.sort(lockList);
                currentLock = lockList.get(0);
                if(currentLock.equals(rootLockPath+"/"+lockName)){
                    this.currentLock = currentLock;
                    return true;
                }else{
                    // 若不是最小节点，则找到自己的前一个节点
                    String prevNode = currentLock.substring(currentLock.lastIndexOf("/") + 1);
                    waitLock = lockList.get(Collections.binarySearch(lockList, prevNode) - 1);
                }
            }else{
                this.currentLock = currentLock;//lockList.get(0);
                return true;
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            logger.error("加锁失败!",e);
            return false;
        }
        return false;
    }

    private boolean lock(final String lockName) {
        return tryLock(lockName,(long)LockConfig.ZOOKEEPER_CONNECTION_TIMEOUT);
    }


}
